/*********************************************************************************
 *      Copyright:  (C) 2020 makun<1394987689@qq.com>
 *                  All rights reserved.
 *
 *       Filename:  mosquitto_pub.c
 *    Description:  This file 
 *                 
 *        Version:  1.0.0(2020年07月08日)
 *         Author:  makun <1394987689@qq.com>
 *      ChangeLog:  1, Release initial version on "2020年07月08日 22时30分04秒"
 *                 
 ********************************************************************************/


#include <stdio.h>
#include <errno.h>
#include <unistd.h>
#include <libgen.h>
#include <getopt.h>
#include <string.h>
#include <mosquitto.h>
#include <stdlib.h>
#include <sys/types.h>
#include <signal.h>
#include <pthread.h>

#include "sqlite3.h"
#include "sample.h"
#include "logger.h"
#include "mosquitto.h"
#include "database.h"
#include "mqtt_config.h"

#include "cJSON.h"


#define  PATH "./aliyun.ini"


int g_stop = 0;
void sig_stop(int signum)//如果传的参数是信号1，SIGUSR1则停止循环
{
	if(SIGUSR1 == signum)
	{
		g_stop = 1;
	}
}

int thread_start_init( pthread_attr_t *thread_attr );
void *thread_pub_work(void *mqtt);
void *thread_sub_work(void *mqtt);

void my_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos);
void my_message_callback( struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg );
void my_connect_callback(struct mosquitto *mosq, void *obj, int rc);
void my_disconnect_callback( struct mosquitto *mosq, void *obj, int rc );




static void  print_usage( char *progname)
{
	printf("Usage:%s [option]...\n", progname);

	printf("%s is a socket server program, which used to verify client and echo back string from it\n", progname);
	printf("\n Mandatory arguments to long options are mandatory for short options too:\n");
	printf("-b(daemon)set progname runing on background\n");
 	printf("-p(--port):sepcify server listen port.\n");
 	printf("-s(--sn):sepcify name.\n");
 	printf("-n(--brokeraddress):sepcify host name.\n");
 	printf("-t(--sample_time):sepcify time to sample  .\n");
	printf("-h(--hellp):print this help information.\n");

	printf("\n Example:%s -b -p 8900\n", progname);
	return ;

}

int main (int argc, char **argv)
{
	int                      daemon_run=0;
	char                    *brokeraddress;
 	int                      sample_time;
	char                    *sn=NULL;
	int                      port;
	float                    temp_data;

    int                      rv = -1;
    int                      opt= -1;
    char                    *progname = NULL;
	pthread_attr_t       thread_attr;
    mqtt_user_data_t         mqtt;
	pthread_t                tid;

    struct mosquitto *mosq = NULL;

	signal(SIGUSR1, sig_stop);//将sigusr1作为参数传给函数sig_stop
	struct option                    long_options[] =
	{

		{"dameon", no_argument, NULL, 'b'},
		{"sample_time", required_argument,NULL, 't'},
		{"port", required_argument, NULL, 'p'},
		{"sn", required_argument, NULL, 'S'},
		{"help", no_argument, NULL, 'h'},
		{"brokeraddress",required_argument, NULL, 'n'},

		{ NULL, 0, NULL, 0}

	};
    

    progname = (char *)basename(argv[0]);

while((opt = getopt_long(argc, argv,"bp:s:t:n:h", long_options, NULL)) != -1)//(需要参数再加:);
	{
		switch(opt)
		{
			case 'b':
				daemon_run=1;
				break;

		
			case 'p':
				port=atoi(optarg);//string -> int
				break;
                        
			case 's':
				sn=optarg;
				break;         

			case 'n':
				brokeraddress=optarg;
				break;

			case 't':
		
				sample_time=atoi(optarg);
				break;
			case 'h':
				print_usage(progname);
				return EXIT_SUCCESS;

			default:
				break;
		}
	}
	if( !port )
	{
		print_usage(progname);
		return -1;
	}

	if(daemon_run)//设置程序背景运行
	{
		daemon(0, 0);
	}

	if(logger_init("stdout", LOG_LEVEL_INFO) <0)
	{
		fprintf(stderr, "inital logger system failure:%s\n", strerror(errno));	
		return -1;
	}
	log_info("open database successfullly!\n");


	if (get_config(PATH, &mqtt)!=0)
	{
		log_error("get_config() failure!\n");
		return -1;
	}
	log_info("confige successfully!\n");

#if 1
	printf("brokeraddress:%s\n", mqtt.brokeraddress);
	printf("port:%d\n", mqtt.port);
	printf("username:%s\n", mqtt.username);
	printf("passwd:%s\n", mqtt.passwd);
	printf("clientid:%s\n", mqtt.clientid);
#endif

    /*必须在任何其他mosquitto功能之前调用*/
    mosquitto_lib_init();


	if( thread_start_init( &thread_attr ) < 0 )   //线程初始化
	{
		log_error( "Thread  start init failure:%s\n", strerror(errno) );
		goto cleanup;
	}

	if( pthread_create( &tid, &thread_attr, thread_pub_work, (void *)&mqtt ) < 0 )  //创建pub线程
	{
		log_error(" Create publish pthread failure:%s\n", strerror(errno) );
		goto cleanup;
	}

	if( pthread_create( &tid, &thread_attr, thread_sub_work, (void *)&mqtt ) < 0 )  
	{
		log_error( "Create subscript pthread failure:%s\n", strerror(errno) );
		goto cleanup;
	}


	while(!g_stop)
	{
		sleep(1);
	}



cleanup:
	
    mosquitto_destroy(mosq);

	pthread_exit(NULL);
    mosquitto_lib_cleanup();

    return 0;

}



void *thread_pub_work(void *mqtt)
{

    bool                     session = true;
	int                      mid=0;
	int                      last_time;
    float                    temp_data;
    char                     time_data[1024];
    int                      sample_flag ;
	int                      connect_flag=0;
	int                      err;
	struct mosquitto        *mosq = NULL;

	mqtt_user_data_t        *mqtt_args;
	st_data                  sample_data;
	char                     data_buff[1024];

	memset(&sample_data, 0, sizeof(sample_data));
	mqtt_args=(mqtt_user_data_t *)mqtt;

    /*创建一个新的mosquitto客户端实例,第二个参数为true，代理清除断开连接时的所有消息和订阅*/
    
	mosq = mosquitto_new(mqtt_args->clientid, session, (void *)&mqtt_args );
    if(!mosq)
    {
        printf("mosquitto new failure: %s\n", strerror(errno));
        goto cleanup;
    }

    printf("pub:Create mosquitto successfuly\n");

    /*设置连接回调，当代理发送CONNACK消息以响应连接时，将调用此方法*/

    /*配置mosquitto实例的用户名和密码*/


	if(mosquitto_username_pw_set(mosq, mqtt_args->username, mqtt_args->passwd ) != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto username and passwd failure:%s\n",strerror(errno));
        goto cleanup;
    }
	printf("pub:mosquitto_username_pw_set ()successfully\n");

	last_time=time(NULL);
	while ( !g_stop)
	{
		sample_flag = 0;

		if((time(NULL) -last_time)>=mqtt_args->sample_time)
		{
                /*采样温度 */
			log_info("start sampling!\n");
			/* sample_data初始化*/
			if (get_temp(&sample_data.temp) != 0)
			{
				log_error("get temperature failure !: %s\n", strerror(errno));

				goto cleanup;
			}
        	get_time( sample_data.time);//获取时间,传buff传大小
			
			strncpy(sample_data.sn, mqtt_args->serial_no, sizeof(sample_data.sn));

/*将获取的数据以json形式打包*/
			if(pack_json(&sample_data, data_buff) < 0)   //打包数据
			{
				log_error("Pack json failure:%s\n", strerror(errno));
				goto cleanup;
			}
			printf("pack_json() successfully!\n");
	    	
			sample_flag = 1;
			last_time = time(NULL);
		}

		if(!connect_flag)   //未连接服务器则连接服务器
		{
			if(mosquitto_connect(mosq, mqtt_args->brokeraddress, mqtt_args->port, mqtt_args->keep_alive) != MOSQ_ERR_SUCCESS)   //连接服务器失败
			{
				printf("Mosquitto connect failure:%s\n", strerror(errno));
				connect_flag = 0;  //将连接标志设为断线
				continue;
    		}

			if(mosquitto_loop_start(mosq) != MOSQ_ERR_SUCCESS)
        	{
            	log_error("Mosquitto loop failure:%s\n", strerror(errno));
    	    	goto cleanup;
			}

			connect_flag = 1;   //连接成功
		}

		if(sample_flag)   //如果已经采样
		{
			printf("publis:%s\n", mqtt_args->pub_topic);

			
			err=mosquitto_publish(mosq, &mid, mqtt_args->pub_topic, strlen(data_buff)+1, data_buff, 0, true);
			if(err)//发布数据
       		{
        		log_error("err=%d Mosquitto Publish failure:%s\n",err, strerror(errno));

				if(mosquitto_loop_stop(mosq, true) != MOSQ_ERR_SUCCESS)    //关闭创建的进程
				{
					log_error("Mosquitto loop stop failure:%s\n", strerror(errno));
					goto cleanup;
				}

				connect_flag = 0;  //将标志设为断线
				continue;
    		}

			log_info("Mosquitto Publish data successfully!\n\n");

		}
	}

cleanup:
	mosquitto_destroy(mosq);
	pthread_exit(NULL);

}



int thread_start_init( pthread_attr_t *thread_attr )
{
	int            rv = -1;
 
	if( pthread_attr_init(thread_attr) )
 	{
 		log_error( "Pthread attr init failure:%s\n", strerror(errno) );
 		goto cleanUp;
 	}
 
	if( pthread_attr_setstacksize(thread_attr, 120*1024) )
 	{
 		log_error( "Pthread attr setstacksize failure:%s\n", strerror(errno) );
 		goto cleanUp;
 	}
 
	if( pthread_attr_setdetachstate( thread_attr, PTHREAD_CREATE_DETACHED) )
 	{
 		log_error( "Pthread attr setdetachstate failure:%s\n", strerror(errno) );
 		goto cleanUp;
 	}
	
	rv = 0;

cleanUp:	
	return rv;
}


void *thread_sub_work(void *mqtt)
{

    bool                     session = true;
	mqtt_user_data_t        *mqtt_args;
	struct mosquitto        *mosq = NULL;

	mqtt_args=(mqtt_user_data_t *)mqtt;

#if 0

	printf("brokeraddress:%s\n", mqtt_args->brokeraddress);
	printf("port:%d\n", mqtt_args->port);
	printf("username:%s\n", mqtt_args->username);
	printf("passwd:%s\n", mqtt_args->passwd);
	printf("clientid:%s\n", mqtt_args->clientid);
	printf("sublish:%s\n", mqtt_args->sub_topic);
#endif

    /*创建一个新的mosquitto客户端实例,第二个参数为true，代理清除断开连接时的所有消息和订阅*/
    mosq = mosquitto_new(mqtt_args->clientid, session, (void *)&mqtt_args );
//    mosq = mosquitto_new("h9fpJz8hFw8.shumei_pi|securemode=2,signmethod=hmacsha256,timestamp=1654438311025|", session, mqtt );
    if(!mosq)
    {
        printf("mosquitto new failure: %s\n", strerror(errno));
        goto cleanup;
    }

    log_info("sub:Create mosquitto successfuly\n");
                                                                                        
   if(mosquitto_username_pw_set(mosq, mqtt_args->username, mqtt_args->passwd ) != MOSQ_ERR_SUCCESS)
//	if(mosquitto_username_pw_set(mosq, "shumei_pi&h9fpJz8hFw8", "73be1400f20de0277b7fee0d81d62e52716673bfe7eeea843663042fe425f673" ) != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto username and passwd failure:%s\n",strerror(errno));
        goto cleanup;
    }
	log_info("sub:mosquitto_username_pw_set ()successfully\n");

	mosquitto_connect_callback_set( mosq, my_connect_callback );
 	mosquitto_disconnect_callback_set( mosq, my_disconnect_callback );
	mosquitto_message_callback_set( mosq, my_message_callback );
    mosquitto_subscribe_callback_set( mosq, my_subscribe_callback );
		
	if(mosquitto_connect(mosq, mqtt_args->brokeraddress, mqtt_args->port, mqtt_args->keep_alive) != MOSQ_ERR_SUCCESS)
	{
		log_error("mosquitto connect server failure:%s\n", strerror(errno));
		goto cleanup;
		
	}
	log_info("sub:mosquitto_connect ()successfully\n");

#if 0
	printf("working here!\n");
		if(mosquitto_loop_forever(mosq, -1, 1) != MOSQ_ERR_SUCCESS)
		{
			log_error("Mosquitto loop_forever() failure:%s\n", strerror(errno));
			goto cleanup;
		}
	printf("2,working here!\n");
#endif
#if 1
	
		
	if(mosquitto_loop_start(mosq) != MOSQ_ERR_SUCCESS)
	{
		log_error("Mosquitto loop failure:%s\n", strerror(errno));
		goto cleanup;
	}

	while( !g_stop )//长连接
	{
		mosquitto_loop_stop( mosq, false );
		pthread_exit(NULL);
	}
#endif
cleanup:
	mosquitto_destroy(mosq);
	pthread_exit(NULL);

}


void my_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
	int               mid = 0;
	mqtt_user_data_t *mqtt;

	mqtt = (mqtt_user_data_t *)obj;
	
	log_info("%d,connect callback!!\n", rc);
	if(!rc)
    {
		printf("sublish:%s\n", mqtt->sub_topic);
        if(mosquitto_subscribe(mosq, &mid, mqtt->sub_topic, 0) != MOSQ_ERR_SUCCESS)
        {
            printf("Mosquitto subscript failure:%s\n", strerror(errno));
            exit(1);
        }

	}
}

void my_message_callback( struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg )
{ 	
	printf("memseage_callback call!\n");
	if( msg->payload )
	{
		if( parse_json(msg->payload) < 0 )
        {	
			printf( "Parse json failure:%s\n", strerror(errno) );
			exit(1);
		}
	}
}

void my_disconnect_callback( struct mosquitto *mosq, void *obj, int rc )
{
	int               mid = 0;
	mqtt_user_data_t *mqtt_args;
	
	mqtt_args = (mqtt_user_data_t *)obj; 

	if(!rc)
	{
	    printf("Mosquitto to server disconnected\n");

		if(mosquitto_connect(mosq, mqtt_args->brokeraddress, mqtt_args->port, mqtt_args->keep_alive) != MOSQ_ERR_SUCCESS)
		{
			printf("Mosquitto connect server failure:%s\n", strerror(errno));
			exit(1);
		}
	}
}

void my_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
    printf("Mosquitto subscribe successfully!\n");
}

